#include <sys/types.h>
#include <sys/mman.h>
#include <sys/stat.h>
#include <unistd.h>
#include <fcntl.h>
#include <semaphore.h>

#include <libgen.h>
#include <errno.h>
#include <stdio.h>
#include <string.h>

#include "queue.h"
#include "log.h"

INI_Section *cfg;

static char data_filename[HANDLER_NUMBER][FILE_PATH_SIZE];
static char data_index   [HANDLER_NUMBER][FILE_PATH_SIZE];

char queue_root[FILE_PATH_SIZE];

queue_control *control;

queue_handler handler[] = {
	{ 0, 0, data_filename[QUEUE_INDEX_HTTP]  , data_index[QUEUE_INDEX_HTTP] },
	{ 0, 0, data_filename[QUEUE_INDEX_SHM]   , data_index[QUEUE_INDEX_SHM]  },
	{ 0, 0, data_filename[QUEUE_INDEX_MSG]   , data_index[QUEUE_INDEX_MSG]  },
	{ 0, 0, data_filename[QUEUE_INDEX_FIFO]  , data_index[QUEUE_INDEX_FIFO] },
	{ 0, 0, data_filename[QUEUE_INDEX_DB]    , data_index[QUEUE_INDEX_DB]   },
	{ 0, 0, data_filename[QUEUE_INDEX_INNER] , data_index[QUEUE_INDEX_INNER]},
	{ 0, 0, data_filename[QUEUE_INDEX_EXTER] , data_index[QUEUE_INDEX_EXTER]}
};

queue_status* status[HANDLER_NUMBER] = {NULL, };

int    rqueue_lock [HANDLER_NUMBER] = {0, };
int is_rqueue_lock [HANDLER_NUMBER] = {0, };

static int is_rqueue_pause[HANDLER_NUMBER] = {0, };

static sem_t rlock[HANDLER_NUMBER];
static sem_t rlock_throw[HANDLER_NUMBER];

int prog_dir(char *prog_path, const char *prog);

int queue_path_init();

void queue_header_init(queue_header *header)
{
	memset(header, 0x0, sizeof(queue_header));
	header->start_sign = 0xFFFFFFFF;
}

int queue_init_conf(const char* config)
{
	char filename[FILE_PATH_SIZE];
	if(config == NULL)
	{
		snprintf(filename, FILE_PATH_SIZE, "%s%s", queue_root, QUEUE_DEFAULT_CONF);
		config = filename;
	}

	if(0 > INI_ReadFile(config, &cfg)) {
		fprintf(stderr, "read confing file failed.\n");
		return -1;
	}
	
	return 1;
}

int queue_path_init()
{
	const char *dir;
	char datapath[FILE_PATH_SIZE];

	/* make queue data path */	
	if(NULL == (dir = INI_GetValue(cfg, "global", "Datadir")))
		dir = QUEUE_DEFAULT_DATA_PATH;

	if(dir[0] != '/') {
		snprintf(datapath, FILE_PATH_SIZE, "%s%s", queue_root, dir);
		dir = datapath;
	}

	if(-1 == access(dir, F_OK) && -1 == mkdir(dir, 0755)) {
		fprintf(stderr, "error while calling mkdir, %s.\n", strerror(errno));
		return -1;
	}

	snprintf(data_filename[QUEUE_INDEX_HTTP], FILE_PATH_SIZE, "%s/%s.%s", dir, QUEUE_DEFAULT_NAME, QUEUE_EXT_HTTP);
	snprintf(data_filename[QUEUE_INDEX_SHM] , FILE_PATH_SIZE, "%s/%s.%s", dir, QUEUE_DEFAULT_NAME, QUEUE_EXT_SHM );
	snprintf(data_filename[QUEUE_INDEX_MSG] , FILE_PATH_SIZE, "%s/%s.%s", dir, QUEUE_DEFAULT_NAME, QUEUE_EXT_MSG );
	snprintf(data_filename[QUEUE_INDEX_FIFO], FILE_PATH_SIZE, "%s/%s.%s", dir, QUEUE_DEFAULT_NAME, QUEUE_EXT_FIFO);
	snprintf(data_filename[QUEUE_INDEX_DB]  , FILE_PATH_SIZE, "%s/%s.%s", dir, QUEUE_DEFAULT_NAME, QUEUE_EXT_DB);
	snprintf(data_filename[QUEUE_INDEX_INNER], FILE_PATH_SIZE, "%s/%s.%s", dir, QUEUE_DEFAULT_NAME, QUEUE_EXT_INNER);
	snprintf(data_filename[QUEUE_INDEX_EXTER], FILE_PATH_SIZE, "%s/%s.%s", dir, QUEUE_DEFAULT_NAME, QUEUE_EXT_EXTER);
	snprintf(data_index[QUEUE_INDEX_HTTP], FILE_PATH_SIZE, "%s/.%s.%s.i", dir, QUEUE_DEFAULT_NAME, QUEUE_EXT_HTTP);
	snprintf(data_index[QUEUE_INDEX_SHM] , FILE_PATH_SIZE, "%s/.%s.%s.i", dir, QUEUE_DEFAULT_NAME, QUEUE_EXT_SHM );
	snprintf(data_index[QUEUE_INDEX_MSG] , FILE_PATH_SIZE, "%s/.%s.%s.i", dir, QUEUE_DEFAULT_NAME, QUEUE_EXT_MSG );
	snprintf(data_index[QUEUE_INDEX_FIFO], FILE_PATH_SIZE, "%s/.%s.%s.i", dir, QUEUE_DEFAULT_NAME, QUEUE_EXT_FIFO);
	snprintf(data_index[QUEUE_INDEX_DB]  , FILE_PATH_SIZE, "%s/.%s.%s.i", dir, QUEUE_DEFAULT_NAME, QUEUE_EXT_DB  );
	snprintf(data_index[QUEUE_INDEX_INNER], FILE_PATH_SIZE, "%s/.%s.%s.i", dir, QUEUE_DEFAULT_NAME, QUEUE_EXT_INNER);
	snprintf(data_index[QUEUE_INDEX_EXTER], FILE_PATH_SIZE, "%s/.%s.%s.i", dir, QUEUE_DEFAULT_NAME, QUEUE_EXT_EXTER);
	/* make queue data path end. */

	/* make queue log path */
	if(-1 == init_log(NULL))
	{
		fprintf(stderr, "init logger error.\n");
		return -1;
	}
	
	return 1;
}

queue_status* stat_open(queue_status **st, const char* filename)
{
	int fd;
	queue_status istat;
	
	if(0 == access(filename, F_OK))
	{
		fd = open(filename, O_RDWR, 0644);
	}
	else
	{
		fd = open(filename, O_RDWR|O_CREAT|O_TRUNC, 0644);
		if(-1 == fd) return MAP_FAILED;
		
		memset(&istat, 0x0, sizeof(queue_status));
		istat.id_next = -1;
		if(-1 == write(fd, &istat, sizeof(queue_status)))
		{
			close(fd);
			return MAP_FAILED;
		}
	}
	
	*st = (queue_status*) mmap(NULL, sizeof(queue_status),
		PROT_READ|PROT_WRITE, MAP_SHARED, fd, 0);

	close(fd);
	return *st;
}

int stat_close(queue_status *st)
{
	return munmap(st, sizeof(queue_status));
}

int queue_status_init()
{
	int i, fd;
	const char *ctl = INI_GetValue(cfg, "global", "ctl");
	if(-1 == queue_path_init())
		return -1;
	
	if(-1 == (fd = shm_open(ctl, O_RDWR, 0644)))
	{
		if(-1 == (fd = shm_open(ctl, O_RDWR|O_CREAT, 0644)))
			return -1;
		ftruncate(fd, sizeof(queue_control));
	}
	
	control = (queue_control*) mmap(NULL, sizeof(queue_control),
		PROT_READ|PROT_WRITE, MAP_SHARED, fd, 0);
	close(fd);
	if(-1 == (int)control) {
		fprintf(stderr, "control mmap get failed in file %s.\n", __FILE__);
		return -1;
	}

	for(i = 0; i < HANDLER_NUMBER; i++)
		if(MAP_FAILED == stat_open(&status[i], handler[i].index))
			return -1;
	return 1;
}

int queue_status_fini(int is_daemon)
{
	int i;
	const char* ctl;
	for(i = 0; i < HANDLER_NUMBER; i++)
		stat_close(status[i]);
	
	munmap(control, sizeof(queue_control));
	if(is_daemon)
	{
		ctl = INI_GetValue(cfg, "global", "ctl");
		shm_unlink(ctl);
	}

	if(-1 == fini_log())
		fprintf(stderr, "close logger error.\n\n");

	INI_Free(cfg);
	return 1;
}

char* root_path(char *prog)
{
	char *ptr;
	if(-1 == prog_dir(queue_root, prog))
		return NULL;

	ptr = queue_root + strlen(queue_root) -1;
	while(*ptr == '/') *ptr-- = '\0';
	while(*ptr != '/') *ptr-- = '\0';
	return queue_root;
}

int prog_dir(char *prog_path, const char *prog)
{
	char* ptr;
	char* rpt;
	size_t size;

	rpt = strrchr(prog, '/');
	size = rpt - prog + 1;
	if(*prog == '/')
	{
		strncpy(prog_path, prog, size);
		prog_path[size] = '\0';
		return 1;
	}

	if(NULL == (ptr = getcwd(prog_path, (size_t)FILE_PATH_SIZE)))
	{
		return -1;
	}

	ptr += strlen(prog_path);
	*ptr++ = '/';
	while(strncmp(prog, "..", 2) == 0)
	{
		prog += 3;
		size -= 3;
		if(*prog == '/')
		{
			prog++;
			size--;
		}

		ptr--;
		while(*ptr == '/') ptr--;
		while(*ptr != '/') ptr--;
		ptr++;
	}

	if(strncmp(prog, "./", 2) == 0)
	{
		prog += 2;
		size -= 2;
	}

	memcpy(ptr, prog, size);
	ptr[size] = '\0';
	return 1;
}

int rqueue_lock_init()
{
	int i;
	for(i = 0; i < HANDLER_NUMBER; i++)
	{
		if(0 != sem_init(&rlock[i], 0, 0))
			return -1;
	}
	
	for(i = 0; i < HANDLER_NUMBER; i++)
	{
		if(0 != sem_init(&rlock_throw[i], 0, 0))
			return -1;
	}
	
	return 1;	
}

void rqueue_lock_fini()
{
	int i;
	for(i = 0; i < HANDLER_NUMBER; i++)
		sem_destroy(&rlock[i]);
	
	for(i = 0; i < HANDLER_NUMBER; i++)
		sem_destroy(&rlock_throw[i]);	
}

int rqueue_read_lock(int index)
{
	int rt;
	
	is_rqueue_lock[index]++;
again:
	if(0 != (rt = sem_wait(&rlock[index])))
	{
		if(errno == EINTR) goto again;
	}

	return rt;
}

int rqueue_read_unlock(int index)
{
	int rt;
	is_rqueue_lock[index]--;
again:
	if(0 != (rt = sem_post(&rlock[index])))
	{
		if(errno == EINTR) goto again;
	}
	
	return rt;
}

int rqueue_read_pause(int index)
{
	int rt;
	if(is_rqueue_pause[index] == 1) return 0;
	is_rqueue_pause[index] = 1;
again:
	if(-1 == (rt = sem_wait(&rlock_throw[index])))
	{
		if(errno == EINTR) goto again;
		return -1;
	}
	
	return 1;
}

int rqueue_read_continue(int index)
{
	int rt;
	if(is_rqueue_pause[index] == 0) return 0;
	
again:
	is_rqueue_pause[index] = 0;
	if(-1 == (rt = sem_post(&rlock_throw[index])))
	{
		if(errno == EINTR) goto again;
		return -1;
	}
	
	rqueue_lock[index] = 0;
	return 1;
}
